/**
 * jmsTemplate三组九种发送消息方法：
 * //将消息发送到指定的Destination
 * public void send(Destination destination, MessageCreator messageCreator)    --发送
 * public void convertAndSend(Destination destination, Object message)    --转换、发送
 * public void convertAndSend(Destination destination, Object message, MessagePostProcessor postProcessor)  --转换、处理、发送
 * //将消息发送到指定的Destination
 * public void send(String destinationName, MessageCreator messageCreator)
 * public void convertAndSend(String destinationName, Object message)
 * public void convertAndSend(String destinationName, Object message, MessagePostProcessor postProcessor)
 * 将消息发送到defaultDestination
 * public void send(MessageCreator messageCreator)
 * public void convertAndSend(Object message)
 * public void convertAndSend(Object message, MessagePostProcessor postProcessor)
 * <p>
 * 与上述send对应的还有一种完全同步发送方法sendAndReceive，在发送后阻塞直到消费者回复或超时才会继续
 * <p>
 * new MessageCreator() {
 * public Message createMessage(Session session) throws JMSException {
 * String text = "A test for MessageCreator";
 * Message message = session.createTextMessage(text);
 * return message;
 * }
 * }
 * <p>
 * new MessagePostProcessor() {
 * public Message postProcessMessage(Message message) throws JMSException {
 * message.setIntProperty("order", 1);
 * return message;
 * }
 * }
 * Connection connection = jmsTemplate.getConnectionFactory().createConnection();
 * connection.start();
 * Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 * <p>
 * * jmsMessagingTemplate有下列发送消息方法：
 * send(Message<?> message)
 * send(String destinationName, Message<?> message)
 * convertAndSend(Object payload)
 * convertAndSend(Object payload, @Nullable MessagePostProcessor postProcessor)
 * convertAndSend(String destinationName, Object payload)
 * convertAndSend(String destinationName, Object payload, @Nullable MessagePostProcessor postProcessor)
 * convertAndSend(String destinationName, Object payload, @Nullable Map<String, Object> headers)
 * convertAndSend(String destinationName, Object payload, @Nullable Map<String, Object> headers,@Nullable MessagePostProcessor postProcessor)
 * 与上述发送方法完全对应的还有一组同步发送方法sendAndReceive和convertSendAndReceive，在发送后阻塞直到消费者回复或超时才会继续
 * Connection connection = jmsMessagingTemplate.getConnectionFactory().createConnection()
 * connection.start();
 * Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 */
package com.example.acitvemqspringbootdemo;

import org.apache.activemq.ScheduledMessage;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.core.*;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Component;


import javax.jms.*;
import java.time.LocalTime;
import java.util.UUID;

@Component
public class Producer {

    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;

    @Autowired
    private JmsTemplate jmsTemplate;
    @Autowired
    private JmsTemplate jmsTemplate_async;
    @Autowired
    private JmsTemplate jmsTemplate_ack;

    @Autowired
    private Queue myQueue;

    @Autowired
    private Topic myTopic;

    @Value("${pub-sub-domain}")
    private Boolean isTopic;

    public void send() {
        try {
            jmsMessagingTemplate.getJmsTemplate().setReceiveTimeout(2000);
            if (isTopic) {
                jmsMessagingTemplate.getJmsTemplate().setPubSubDomain(true);
                System.out.println("Sending topic message......");
                jmsMessagingTemplate.convertAndSend(myTopic, "Simple Message from myTopic...." + System.currentTimeMillis());
            } else {
                jmsMessagingTemplate.getJmsTemplate().setPubSubDomain(false);
                System.out.println("Sending queue message......");

                org.springframework.messaging.Message<String> msg = MessageBuilder.withPayload("Simple Message from myQueue by send")
                        .build();
                System.out.println("[send]sent:" + msg.getPayload() + "-----" + msg.getHeaders());
                jmsMessagingTemplate.send(myQueue.getQueueName(), msg);

                msg = MessageBuilder.withPayload("Simple Message from myQueue by sendAndReceive").build();
                System.out.println("[sendAndReceive]sent:" + msg.getPayload() + "-----" + msg.getHeaders());
                org.springframework.messaging.Message<?> rMessage = jmsMessagingTemplate.sendAndReceive(myQueue.getQueueName(), msg);
                if (rMessage != null)
                    System.out.println("[S&R]reply:" + rMessage.getPayload());

                System.out.println("[convertAndSend]sent:" + "Simple Message from myQueue by convertAndSend....");
                jmsMessagingTemplate.convertAndSend(myQueue, "Simple Message from myQueue by convertAndSend....");

                System.out.println("[convertSendAndReceive]sent:" + "Simple Message from myQueue by convertSendAndReceive....");
                String rString = jmsMessagingTemplate.convertSendAndReceive(myQueue.getQueueName(), "Simple Message from myQueue by convertSendAndReceive....", String.class);
                System.out.println("[C&S&R]reply:" + rString);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Autowired
    @Qualifier("sendQueue")
    private Destination sendDestination;


    private Destination replyDestination;
    @Autowired
    @Qualifier("replyQueue")
    public void setReplyDestination(Destination replyDestination) {
        this.replyDestination = replyDestination;
    }


    public void sendReply(int num) throws Exception {
        //构造接收reply的Listener
        Connection conn = jmsTemplate_async.getConnectionFactory().createConnection();
        conn.start();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        MessageConsumer replyConsumer = session.createConsumer(replyDestination);
        replyConsumer.setMessageListener(new MessageListener() {
            public void onMessage(Message m) {
                try {
                    System.out.println("[Sender]received:" + ((TextMessage) m).getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });

        //发送数据
        jmsTemplate_async.send(sendDestination, new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                String msgId = UUID.randomUUID().toString();
                TextMessage message = session.createTextMessage("sending message....");
                message.setJMSCorrelationID(msgId);
                message.setIntProperty("listener", num);
//              message.setJMSReplyTo(replyDestination);         //设置回复队列地址
                return message;
            }
        });
    }

    public void sendReceive() throws Exception {
        jmsTemplate.setReceiveTimeout(2000);
        Message received = jmsTemplate.sendAndReceive("send_receive_queue", new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                String msgId = UUID.randomUUID().toString();
                TextMessage message = session.createTextMessage("foo");
                message.setJMSCorrelationID(msgId);
                return message;
            }
        });
        if (received != null)
            System.out.println("[sendAndReceive]received reply: " + ((TextMessage) received).getText());
        else
            System.out.println("[sendAndReceive]received reply: Timeout");
    }

    public void transactionProducer() {
        String msg = "Spring Transaction Message..."+LocalTime.now();
        System.out.println("[SpringTransaction]sending：" + msg);
        //启用事务
        jmsTemplate.setSessionTransacted(true);      //默认为false
//        jmsTemplate.setSessionAcknowledgeMode(4);
        jmsTemplate.convertAndSend(new ActiveMQQueue("springTransactionQueue"), msg);
    }

    public void durableProducer() {
        String msg = "Spring Durable Message..." + LocalTime.now();
        System.out.println("[SpringDurable]sent：" + msg);
        //启用持久化（缺省）
        jmsTemplate.setExplicitQosEnabled(true);   //deliveryMode, priority, timeToLive 的开关，要生效，必须配置为true，默认false
        jmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT);
        if (!isTopic) {
            jmsTemplate.convertAndSend(new ActiveMQQueue("springDurableQueue"), msg);
        }
        else{
             jmsTemplate.setPubSubDomain(true);
             jmsTemplate.convertAndSend(new ActiveMQTopic("springDurableTopic"), msg);
             jmsTemplate.setPubSubDomain(false);
        }
    }

    //消息手动确认，需要在JmsTemplate中关闭事务和设定消息确认方式
    public void ackProducer() {
        for (int i = 0; i < 5; i++) {
            String msg = "Spring Ack Message..." + (i + 1);
            System.out.println("[springAck]sending：" + msg);
            jmsTemplate_ack.convertAndSend(new ActiveMQQueue("springAckQueue"), msg);
        }
    }

    //死信发送
    public void deadProducer() {
        String msg = "Spring dead Message..."+LocalTime.now();
        jmsTemplate.setSessionTransacted(true);
        jmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT);
        jmsTemplate.convertAndSend(new ActiveMQQueue("springDeadQueue"),msg);
        System.out.println("[Spring Dead]sent:" + msg);
    }

    //延迟发送，首先在activemq.xml中添加 schedulerSupport="true"
    public void delayProducer() {
        jmsTemplate.convertAndSend(new ActiveMQQueue("springDelayQueue"), "springDelayMessage", m -> {
            m.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 2000);
            return m;
        });
        System.out.println("[Spring Delay]sent:" + "---" + LocalTime.now());
    }

   //利用回调使用Session
    public void sendBySessionCallback(){
       jmsTemplate.execute( new SessionCallback<Object>() {
           public Object doInJms(Session session) throws JMSException {
               MessageProducer producer = session.createProducer(myQueue);
               String msg = "message send by SeesionCallBack";
               TextMessage message = session.createTextMessage(msg);
               producer.send(message);
               System.out.println("[SessionCallback]sent" + msg);
               return null;
           }
       });
    }

    //利用回调使用Session和Producer
    public void sendByProducerCallback(){
     jmsTemplate.execute( new ProducerCallback<Object>(){
            public Object doInJms(Session session, MessageProducer producer) throws JMSException {
                String msg="message send by ProducerCallBack";
                TextMessage message=session.createTextMessage(msg);
                producer.send(myQueue,message);
                System. out .println( "[ProducerCallback]sent:"+msg );
                return null ;
            }
        });
    }
}
